#pragma once
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>

#include "Channel.hpp"
#include "Common/thread_pool.hpp"
#include "Comsumer.hpp"
#include "MQ_Virtual_Engine/VirtualHost.hpp"
#include "codec/codec.h"
#include "fmtlog/fmtlog.h"
#include "mq_proto.pb.h"
#include "muduo/net/Callbacks.h"

namespace MY_MQ {
class Connection {
public:
    using self_ptr = std::shared_ptr<Connection>;

    Connection(const VirtualHost::self_ptr &host, const ConsumerManager::self_ptr &cmp, const ProtobufCodecPtr &codec,
               const muduo::net::TcpConnectionPtr &conn, const Message_MQ::ThreadPool::self_ptr &pool)
        : _conn(conn),
          _codec(codec),
          _Cmp(cmp),
          _host(host),
          _TPool(pool),
          _channelptr(std::make_shared<ChannelManager>()) {
    }

    void OpenChannel(const openChannelRequestPtr &req) {
        bool ret = _channelptr->OpenChannel(req->cid(), _host, _Cmp, _TPool, _codec, _conn);
        if (false == ret) {
            logd("信道重复创建");
            return basicResponse(false, req->rid(), req->cid());
        }
        logd("{}信道创建成功", req->cid());
        return basicResponse(1, req->rid(), req->cid());
    }

    void closeChannel(const closeChannelRequestPtr &req) {
        _channelptr->CloseChannel(req->cid());
        return basicResponse(1, req->rid(), req->cid());
    }

    Channel::self_ptr getChannel(const std::string &cid) {
        return _channelptr->getChannel(cid);
    }

private:
    void basicResponse(bool ok, const std::string &rid, const std::string &cid) {
        MQ_Protobuf::basicCommonResponse resp;
        resp.set_rid(rid);
        resp.set_cid(cid);
        resp.set_ok(ok);
        _codec->send(_conn, resp);
    }

private:
    muduo::net::TcpConnectionPtr     _conn;
    ProtobufCodecPtr                 _codec;
    ConsumerManager::self_ptr        _Cmp;
    VirtualHost::self_ptr            _host;
    Message_MQ::ThreadPool::self_ptr _TPool;
    ChannelManager::self_ptr         _channelptr;
};

class ConnectionManager {
public:
    using self_ptr = std::shared_ptr<ConnectionManager>;

    void newConnection(const VirtualHost::self_ptr &host, const ConsumerManager::self_ptr &CmP, ProtobufCodecPtr codec,
                       const muduo::net::TcpConnectionPtr &conn, const Message_MQ::ThreadPool::self_ptr &Tpool) {
        std::lock_guard lock(_mutex);
        auto            it = _conns.find(conn);
        if (_conns.end() != it) {
            return;
        }
        Connection::self_ptr self_conn = std::make_shared<Connection>(host, CmP, codec, conn, Tpool);
        _conns[conn]                   = self_conn;
    }

    void DeleteConnection(const muduo::net::TcpConnectionPtr &conn) {
        std::lock_guard lock(_mutex);
        _conns.erase(conn);
    }

    Connection::self_ptr getConnection(const muduo::net::TcpConnectionPtr &conn) {
        std::lock_guard lock(_mutex);
        auto            it = _conns.find(conn);
        if (_conns.end() == it) {
            return nullptr;
        }
        return it->second;
    }

private:
    std::mutex                                                             _mutex;
    std::unordered_map<muduo::net::TcpConnectionPtr, Connection::self_ptr> _conns;
};
};  // namespace MY_MQ